package com.zq.learn.springbatch.combination.batch;

import com.zq.learn.springbatch.mongo.StatisticsResult;
import com.zq.learn.springbatch.mongo.sourcedata.SourceData;
import org.quartz.Job;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.item.data.MongoItemReader;
import org.springframework.batch.item.data.MongoItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * Task2
 *
 * @author ZhangQi
 * @date 2020/2/19
 */
@Component
public class OrderTask implements Job {
  private Logger log= LoggerFactory.getLogger(OrderTask.class);
  @Autowired
  private JobBuilderFactory jobBuilderFactory;
  @Autowired
  private StepBuilderFactory stepBuilderFactory;
  @Autowired
  private MongoItemReader<SourceData> orderItemReader;
  @Autowired
  private MongoItemWriter<StatisticsResult> orderItemWriter;
  @Autowired
  private JobLauncher jobLauncher;

  @Override public void execute(JobExecutionContext jobExecutionContext)
    throws JobExecutionException {
    JobDataMap dataMap = jobExecutionContext.getMergedJobDataMap();
    String day = dataMap.getString("day");
    Long status=1L;
    log.info("===========参数：day={},status={}",day,status);
    //统计金额Step
    Step priceStep =
      this.stepBuilderFactory.get("OrderPriceStep_"+day)
          .<SourceData, StatisticsResult>chunk(10)
          .reader(orderItemReader)
          .processor(new OrderPriceProcessor())
          .writer(orderItemWriter)
          .build();
    //统计数量Step
    Step countStep =
      this.stepBuilderFactory.get("OrderCountStep_"+day)
        .<SourceData, StatisticsResult>chunk(10)
        .reader(orderItemReader)
        .processor(new OrderCountProcessor())
        .writer(orderItemWriter)
        .build();
    //创建统计任务Job（
    org.springframework.batch.core.Job job =
      jobBuilderFactory.get("orderJob").start(countStep).next(priceStep).build();
    JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();
    jobParametersBuilder.addLong("status",status);
    jobParametersBuilder.addString("day",day);
    jobParametersBuilder.addLong("timestamp",System.currentTimeMillis());
    try {
      jobLauncher.run(job,jobParametersBuilder.toJobParameters());
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
}
